package com.mashibing.flinkscala.code.chapter6.kryoser

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

import scala.collection.mutable.ListBuffer

/**
 * Scala 序列化
 */
object SerializerTest {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //自定义注册kryo序列化
    env.getConfig.registerTypeWithKryoSerializer(classOf[Student],classOf[StudentSerializer])

    //导入隐式转换
    import org.apache.flink.streaming.api.scala._

    val ds: DataStream[String] = env.socketTextStream("mynode5", 9999)

    ds.map(one=>{
      val arr: Array[String] = one.split(",")
      new Student(arr(0).toInt,arr(1),arr(2).toInt)
    }).print()

    env.execute()
  }

}
